package promote.brozen.reactor;

import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Operators;
import reactor.core.publisher.UnicastProcessor;

/**
 * @author Brozen
 * @date 2020/6/26 9:56 PM
 * @email brozen@qq.com
 */
public class FluxSwitchOnFirst {

    public static void main(String[] args) {
        testRSocketChannel();
    }

    private static void testRSocketChannel() {
        Flux<Object> flux = Flux.from(subscriber -> {
            subscriber.onNext("1");
            subscriber.onNext("2");
            subscriber.onNext("3");
        });


        flux.switchOnFirst((signal, f) -> {
            final UnicastProcessor<String> receiver = UnicastProcessor.create();

            return receiver.transform(Operators.<String, String>lift((scannable, actual) -> new BSubscriber(actual)));
        }).subscribe(new BaseSubscriber<String>() {
            @Override
            protected void hookOnNext(String value) {
                System.out.println(value);
            }
        });
    }

    private static void testSimple() {
        Flux<String> stringFlux = Flux.just("start").concatWith(subscriber -> {
            subscriber.onNext("1");
            subscriber.onNext("2");
            subscriber.onNext("3");
        });

        stringFlux.switchOnFirst((signal, flux) -> {
            String value = signal.get();
            System.out.println("Got from signal: " + value);
            return Flux.just("a", "b", "c");
        }).subscribe(System.out::println);
    }

}
